package com.cti.commons.delayqueue.core;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.cti.commons.delayqueue.core.mode.DelayMessage;
import com.cti.commons.delayqueue.core.mode.ScoredSortedItem;
import com.cti.commons.delayqueue.repository.DelaySendMessageLog;
import com.cti.commons.delayqueue.repository.DelaySendMessageRepository;

/**
 * @author et.wang
 * @description: 延迟消息队列
 * @date 2022/10/27 19:35
 */
public class DelayQueue {

    public static final int SIZE = 1024;

    private static final Logger logger = LoggerFactory.getLogger(DelayQueue.class);

    @Autowired
    private DelayBucket delayBucket;

    @Autowired
    private DelayQueueJobPool delayQueueJobPool;

    @Autowired
    private ReadyQueue readyQueue;

    @Autowired
    private DelaySendMessageRepository repository;

    /**
     * 获取delayBucket key 分开多个，有利于提高效率
     *
     * @param delayQueueJobId
     * @return
     */
    private String getDelayBucketKey(long delayQueueJobId) {
        return DelayQueueConstant.DELAY_BUCKET_KEY_PREFIX + Math.floorMod(delayQueueJobId, DelayQueueConstant.DELAY_BUCKET_NUM);
    }

    /**
     * 添加延迟任务到延迟队列
     *
     * @param delayMessage
     */
    public boolean push(DelayMessage delayMessage) {
        if (delayMessage.getBody() == null) {
            return false;
        }
        DelaySendMessageLog log = new DelaySendMessageLog(delayMessage.getId() + "", delayMessage.getTopic(),
            delayMessage.toString(), null);
        try {
            delayQueueJobPool.addDelayQueueJob(delayMessage);
            ScoredSortedItem item = new ScoredSortedItem(delayMessage.getId(),
                delayMessage.getDelayTime(), delayMessage.getTopic());
            String delayBucketKey = getDelayBucketKey(delayMessage.getId());
            delayBucket.addToBucket(delayBucketKey, item);
            log.success();
            repository.save(log);
            logger.info("DelayQueue Push 发送信息：{},桶信息:{},{}",
                JSON.toJSONString(delayMessage),delayBucketKey,JSON.toJSONString(item));
        } catch (Exception e) {
            logger.error("DelayQueue Push 消息发送失败！", e);
            log.fail();
            String errorMessageString = e.getMessage();
            log.setException(errorMessageString.length() < SIZE ? errorMessageString :
                errorMessageString.substring(0, SIZE));
            repository.save(log);
            return false;
        }
        return true;
    }

    /**
     * 获取准备好的延迟任务
     *
     * @param topic
     * @return
     */
    public DelayMessage pop(String topic) {
        Long delayQueueJobId = readyQueue.pollFormReadyQueue(topic);
        if (delayQueueJobId == null) {
            return null;
        } else {
            DelayMessage delayMessage = delayQueueJobPool.getDelayQueueJob(delayQueueJobId);
            if (delayMessage == null) {
                return null;
            } else {
                long delayTime = delayMessage.getDelayTime();
                //获取消费超时时间，重新放到延迟任务桶中
                long reDelayTime = System.currentTimeMillis() + delayMessage.getTtrTime() * 1000L;
                delayMessage.setDelayTime(reDelayTime);
                delayQueueJobPool.addDelayQueueJob(delayMessage);
                ScoredSortedItem item = new ScoredSortedItem(delayMessage.getId(),
                    reDelayTime, delayMessage.getTopic());
                delayBucket.addToBucket(getDelayBucketKey(delayMessage.getId()), item);
                //返回的时候设置回
                delayMessage.setDelayTime(delayTime);
                return delayMessage;
            }
        }
    }

    /**
     * 删除延迟队列任务
     *
     * @param
     */
    public void delete(long messageId) {
        delayQueueJobPool.deleteDelayQueueJob(messageId);
    }

    /**
     * 删除队列和桶里任务
     *
     * @param messageId
     */
    public void finish(long messageId) {
        DelayMessage delayMessage = delayQueueJobPool.getDelayQueueJob(messageId);
        if (delayMessage == null) {
            return;
        }
        delayQueueJobPool.deleteDelayQueueJob(messageId);
        ScoredSortedItem item = new ScoredSortedItem(delayMessage.getId(),
            delayMessage.getDelayTime(), delayMessage.getTopic());
        delayBucket.deleteFormBucket(getDelayBucketKey(delayMessage.getId()), item);
    }

    /**
     * 查询delay job
     *
     * @param messageId
     * @return
     */
    public DelayMessage get(long messageId) {
        return delayQueueJobPool.getDelayQueueJob(messageId);
    }

}
